package com.bujunjie.study.realtime.dwd.db.split.function;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.bujunjie.study.realtime.common.enums.FlinkOpEnum;
import com.bujunjie.study.realtime.common.util.JdbcUtil;
import com.bujunjie.study.realtime.common.model.vo.DwdTableVO;
import com.bujunjie.study.realtime.dim.model.vo.Mysql2MaxwellVO;
import com.bujunjie.study.realtime.dim.model.vo.ProductVO;
import com.google.common.collect.Maps;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

/**
 * <p>dwd 基础表</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/15 10:33
 */
public class DwdTableProcessFunction extends BroadcastProcessFunction<Mysql2MaxwellVO<ProductVO>, DwdTableVO, Tuple2<Map<String, Object>, DwdTableVO>> {

    private MapStateDescriptor<String, DwdTableVO> mapStateDescriptor;

    private Map<String, DwdTableVO> configMap = Maps.newHashMap();

    public DwdTableProcessFunction(MapStateDescriptor<String, DwdTableVO> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        //  项目初始化，获取配置信息
        Connection mySQLConnection = JdbcUtil.getMySQLConnection();
        List<DwdTableVO> dwdTableVOS = JdbcUtil.queryList(mySQLConnection, "select * from gmall2024_config.dwd_table_process", DwdTableVO.class, true);
        for (DwdTableVO dwdTableVO : dwdTableVOS) {
            String sourceTable = dwdTableVO.getSourceTable();
            String sourceType = dwdTableVO.getSourceType();
            String key = this.getKey(sourceTable, sourceType);
            configMap.put(key, dwdTableVO);
        }
    }


    /**
     * <p>生成对应的 key</p>
     *
     * @param sourceTable 原表
     * @param sourceType  原类型
     * @return 返回对应的 key 值
     * @author bu.junjie
     * @date 2025/9/15 10:50
     */
    private String getKey(String sourceTable, String sourceType) {
        return sourceTable + ":" + sourceType;
    }

    @Override
    public void processElement(Mysql2MaxwellVO<ProductVO> value, BroadcastProcessFunction<Mysql2MaxwellVO<ProductVO>, DwdTableVO,
            Tuple2<Map<String, Object>, DwdTableVO>>.ReadOnlyContext ctx, Collector<Tuple2<Map<String, Object>, DwdTableVO>> out) throws Exception {
        String table = value.getTable();
        String type = value.getType();
        String key = this.getKey(table, type);
        ReadOnlyBroadcastState<String, DwdTableVO> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if (broadcastState.contains(key) || configMap.containsKey(key)) {
            ProductVO data = value.getData();
            DwdTableVO dwdTableVO = this.getDwdTableVO(key, broadcastState);
            Map<String, Object> filterMap = this.filterNeedColumns(data, dwdTableVO.getSinkColumns());
            Long ts = data.getTs();
            filterMap.put("ts", ts);
            out.collect(Tuple2.of(filterMap, dwdTableVO));
        }

    }

    /**
     * <p>获取 dwd table值</p>
     *
     * @param
     * @return
     * @author bu.junjie
     * @date 2025/9/15 11:03
     */
    private DwdTableVO getDwdTableVO(String key, ReadOnlyBroadcastState<String, DwdTableVO> broadcastState) throws Exception {
        DwdTableVO dwdTableVO = null;
        dwdTableVO = broadcastState.get(key);
        if (ObjectUtil.isNull(dwdTableVO)) {
            dwdTableVO = configMap.get(key);
        }
        return dwdTableVO;
    }

    @Override
    public void processBroadcastElement(DwdTableVO dwdTable, BroadcastProcessFunction<Mysql2MaxwellVO<ProductVO>, DwdTableVO,
            Tuple2<Map<String, Object>, DwdTableVO>>.Context ctx, Collector<Tuple2<Map<String, Object>, DwdTableVO>> out) throws Exception {
        //  获取配置表的操作类型
        BroadcastState<String, DwdTableVO> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String op = dwdTable.getOp();
        String sourceTable = dwdTable.getSourceTable();
        String sourceType = dwdTable.getSourceType();
        String key = this.getKey(sourceTable, sourceType);
        if (FlinkOpEnum.D.getOp().equals(op)) {
            configMap.remove(key);
            broadcastState.remove(key);
        } else {
            configMap.put(sourceTable, dwdTable);
            broadcastState.put(key, dwdTable);
        }
    }


    /**
     * <p>
     * 过滤出需要的字段
     *     <ol>
     *         <li>{"tm_name":"Redmi","create_time":"2021-12-14 00:00:00","logo_url":"555","id":1}</li>
     *         <li>sinkColumns  id,tm_name</li>
     *     </ol>
     * </p>
     *
     * @param
     * @return 返回复合规范的字段
     * @author bu.junjie
     * @date 2025/9/15 8:54
     */
    private Map<String, Object> filterNeedColumns(ProductVO productVO, String sinkColumns) {
        Map<String, Object> filterMap = Maps.newHashMap();
        List<String> columnList = Arrays.asList(sinkColumns.split(StrUtil.COMMA));
        Map<String, Object> propMap = BeanUtil.beanToMap(productVO, true, false);
        propMap.forEach((key, value) -> {
            if (columnList.contains(key)) {
                filterMap.put(key, value);
            }
        });
        return filterMap;
    }
}
